package com.ygx.flink.practice.sink;

import com.ygx.flink.practice.model.PvUv;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * https://www.jianshu.com/p/9ab82ef94a9c
 *
 * @author YangGuoXiang
 * @version 1.0
 * @date 2022/3/25 17:29
 */
public class SinkToMysql extends RichSinkFunction<PvUv> {
    private Connection connection;
    private PreparedStatement ps;
    private Logger log = LoggerFactory.getLogger(this.getClass());

    private Connection getConnection() {
        Connection conn = null;
        String url = "jdbc:mysql://192.168.5.9:3306/flink-test?characterEncoding=utf8&useSSL=false";
        String user = "root";
        String password = "123456";
        log.info("SinkToMysql getConnection() log");
        try {
            System.out.println("SinkToMysql getConnection() system");
            // 加上后发布到生产环境会报错 null
//            Class.forName("com.mysql.jdbc.Driver");
            conn = DriverManager.getConnection(url, user, password);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return conn;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        connection = getConnection();
        String insertSql = "insert into pvuv_sink(dt, pv, uv) values (?,?,?)";
        ps = connection.prepareStatement(insertSql);
    }

    @Override
    public void invoke(PvUv value, Context context) throws Exception {
        try {
            ps.setString( 1, value.getDt() );
            ps.setInt( 2, value.getPv() );
            ps.setInt( 3, value.getUv() );
            ps.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void close() throws Exception {
        super.close();

        if(ps != null) {
            ps.close();
        }
        if(connection != null) {
            connection.close();
        }
    }
}